package com.icehand.canal.client.transfer;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.icehand.canal.annotation.HandlePoint;
import com.icehand.canal.client.Point;
import com.icehand.canal.config.CanalConfig;
import com.icehand.canal.handler.CanalHandler;
import org.apache.commons.lang3.StringUtils;

import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;

/**
 * @author icehand
 */
public class DefaultMessageTransponderImpl extends AbstractMessageTransponder{
    public DefaultMessageTransponderImpl(CanalConnector canalConnector, CanalConfig.Instance instance, String destination,
                                         List<CanalHandler> handlers, List<Point> points) {
        super(canalConnector, instance, destination, handlers, points);
    }

    @Override
    protected void processDDL(CanalEntry.RowChange rowChange) {
        //do nothing
    }

    @Override
    protected Predicate<Map.Entry<Method, HandlePoint>> getAnnotationFilter(String destination, String schemaName, String tableName, CanalEntry.EventType eventType) {
        Predicate<Map.Entry<Method, HandlePoint>> df = e -> StringUtils.isEmpty(e.getValue().destination())
                || e.getValue().destination().equals(destination);
        Predicate<Map.Entry<Method, HandlePoint>> sf = e -> e.getValue().schema().length == 0
                || Arrays.asList(e.getValue().schema()).contains(schemaName);
        Predicate<Map.Entry<Method, HandlePoint>> tf = e -> e.getValue().table().length == 0
                || Arrays.asList(e.getValue().table()).contains(tableName);
        Predicate<Map.Entry<Method, HandlePoint>> ef = e -> e.getValue().eventType().length == 0
                || Arrays.stream(e.getValue().eventType()).anyMatch(ev -> ev == eventType);
        return df.and(sf).and(tf).and(ef);
    }

    @Override
    protected Object[] getInvokeArgs(Method method, CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        return Arrays.stream(method.getParameterTypes())
                .map(p -> p == CanalEntry.EventType.class
                        ? eventType
                        : p == CanalEntry.RowData.class
                        ? rowData : null)
                .toArray();
    }

}
